package com.aiwiown.snackmq.broker.handler.console;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.metadata.TopicMetadata;
import com.aiwiown.snackmq.common.dto.TopicInfo;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.serialization.JsonSerializer;
import com.aiwiown.snackmq.common.serialization.SerializerFactory;
import lombok.extern.slf4j.Slf4j;

/**
 * 处理来自控制台的“列出所有 Topic”请求。
 * 【核心修复】：此处理器现在实现了“从 Leader 读取”模式，以确保数据一致性。
 */
@Slf4j
public class ListTopicsHandler extends AbstractLeaderRequiredConsoleHandler {

    private final JsonSerializer serializer = SerializerFactory.getJsonSerializer();

    public ListTopicsHandler(MetadataManager metadataManager, AuthorizationService authorizationService) {
        super(metadataManager, authorizationService);
    }

    @Override
    public MessageType getMessageType() {
        return MessageType.LIST_TOPICS_REQUEST;
    }

    @Override
    protected void doHandle(RequestContext ctx, Message request) {
        try {
            java.util.List<TopicMetadata> allTopics = metadataManager.getAllTopics();
            java.util.List<TopicInfo> topicInfos = allTopics.stream()
                    .map(topicMetadata -> {
                        java.util.Map<Integer, TopicInfo.PartitionAssignmentInfo> assignments = topicMetadata.getPartitionAssignments().entrySet().stream()
                                .collect(java.util.stream.Collectors.toMap(
                                        java.util.Map.Entry::getKey,
                                        entry -> {
                                            TopicMetadata.PartitionAssignment pa = entry.getValue();
                                            return new TopicInfo.PartitionAssignmentInfo(pa.getLeader(), pa.getReplicas());
                                        }
                                ));
                        return new TopicInfo(
                                topicMetadata.getPartitionCount(),
                                topicMetadata.getReplicationFactor(),
                                topicMetadata.getTopicName(),
                                assignments
                        );
                    })
                    .collect(java.util.stream.Collectors.toList());
            byte[] body = serializer.serialize(topicInfos);
            Message response = Message.builder()
                    .type(MessageType.RESPONSE)
                    .messageId(request.getMessageId())
                    .status(MessageStatus.SUCCESS)
                    .body(body)
                    .build();
            ctx.writeResponse(response);
        } catch (Exception e) {
            log.error("Error processing list topics request", e);
            Message response = Message.builder()
                    .type(MessageType.RESPONSE)
                    .messageId(request.getMessageId())
                    .status(MessageStatus.FAILED)
                    .body(("Failed to list topics: " + e.getMessage()).getBytes())
                    .build();
            ctx.writeResponse(response);
        }
    }
}